package com.haoziqi.chapter_05;

import org.apache.flink.streaming.api.datastream.ConnectedStreams;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * description
 * created by A on 2021/3/5
 */
public class ConnectAndUnion {
    public static void main(String[] args) {
        //1.创建环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<Integer> intStream = env.fromElements(1, 2, 3, 4, 5);
        DataStreamSource<String> stringStream = env.fromElements("a", "b", "c");
        DataStreamSource<String> stringStream1 = env.fromElements("a", "b", "c");
        stringStream.union(stringStream1).print("使用union连接的两个算子类型必须一样");
        ConnectedStreams<Integer, String> connect = intStream.connect(stringStream);
        DataStream<Integer> firstInput = connect.getFirstInput();
        DataStream<String> secondInput = connect.getSecondInput();
        firstInput.print("firstEle");
        secondInput.print("secondEle");

        //提交任务
        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
